package org.qiyu.live.im.core.server.handler.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandlerContext;
import jakarta.annotation.Resource;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.qiyu.live.common.interfaces.enums.ImCoreServerIPRedisKey;
import org.qiyu.live.im.constants.ImConstants;
import org.qiyu.live.im.core.server.common.*;
import org.qiyu.live.im.core.server.handler.SimpleHandler;
import org.qiyu.live.im.dto.ImMsgBody;
import org.qiyu.live.im.interfaces.ImTokenRpc;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.qiyu.live.im.core.server.dto.ImOnlineDTO;

import java.util.concurrent.TimeUnit;

import static org.qiyu.live.common.interfaces.topic.ImCoreServerProviderTopicNames.IM_ONLINE_TOPIC;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-07
 * Time: 16:47
 */
@Component
public class LoginMsgHandler implements SimpleHandler {

    private final Logger LOGGER = LoggerFactory.getLogger(LoginMsgHandler.class);

    @DubboReference
    private ImTokenRpc imTokenRpc;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private MQProducer mqProducer;


    /**
     * 接入IM服务后,进行校验token, 登陆handler实现
     *
     * @param ctx
     * @param imMsg
     */
    @Override
    public void handler(ChannelHandlerContext ctx, ImMsg imMsg) {
        // 防止重复发送登录的数据包
        if (ImContextUtils.getUserId(ctx) != null) {
            LOGGER.error("用户重复登录");
            return;
        }

        byte[] body = imMsg.getBody();
        // 1. 对body进行校验
        if (body == null || body.length == 0) {
            ctx.close();
            LOGGER.error("body error msg={}", imMsg);
            throw new IllegalArgumentException("body error");
        }

        // 2. 获取到body的信息(得到ImMsgBody对象),对其进行校验
        ImMsgBody imMsgBody = JSON.parseObject(body, ImMsgBody.class);
        Long userId = imMsgBody.getUserId();
        int appId = imMsgBody.getAppId();
        if (userId == null || userId < 10000 || appId < 10000) {
            ctx.close();
            LOGGER.error("userId or appId  error msg={}", imMsg);
            throw new IllegalArgumentException("userId or appId  error");
        }

        // 3. 校验token,token合法,代表用户登陆成功,回传消息给客户端
        String token = imMsgBody.getToken();
        Long userIdByToken = imTokenRpc.getUserIdByToken(token);

        if (!StrUtil.isEmpty(token) && userIdByToken != null && userIdByToken.equals(userId)) {
            loginSuccess(ctx, userId, appId, token);
            // 我们进行监听用户上线,发送一个MQ消息->将roomId与userId进行绑定
            sendLoginMQ(userId, appId, null);
            return;
        }

        // 4. token不合法,那么直接关闭连接,报错,提示
        LOGGER.error("token error msg={}", imMsg);
        ctx.close();
        throw new IllegalArgumentException("token error");
    }

    /**
     * 监听用户上线,发送消息给MQ
     *
     * @param userId 用户id
     * @param appId  产品id
     * @param roomId 直播间id
     */
    private void sendLoginMQ(Long userId, Integer appId, Integer roomId) {
        ImOnlineDTO imOnlineDTO = new ImOnlineDTO();
        imOnlineDTO.setLoginTime(System.currentTimeMillis());
        imOnlineDTO.setRoomId(roomId);
        imOnlineDTO.setUserId(userId);
        imOnlineDTO.setAppId(appId);
        Message message = new Message();
        message.setTopic(IM_ONLINE_TOPIC);
        message.setBody(JSON.toJSONBytes(imOnlineDTO));
        try {
            SendResult sendResult = mqProducer.send(message);
            LOGGER.info("发送消息成功 res is {}", sendResult);
        } catch (Exception e) {
            LOGGER.error("[sendLoginMQ] 发送消息失败", e);
        }
    }

    private void loginSuccess(ChannelHandlerContext ctx, Long userId, int appId, String token) {
        // token校验成功,
        // 1. 保存该连接,并且设置相关业务属性,方便后续使用
        ChannelHandlerContextCache.put(userId, ctx);

        ImContextUtils.setUserId(userId, ctx);
        ImContextUtils.setAppId(appId, ctx);

        // 2. 回写消息给客户端
        ImMsgBody msgBody = new ImMsgBody();
        msgBody.setAppId(appId);
        msgBody.setUserId(userId);
        msgBody.setToken(token);
        msgBody.setData("true");
        ImMsg returnMsg = ImMsg.build(ImMsgCodeEnum.IM_LOGIN_MSG.getCode(), JSON.toJSONString(msgBody));

        // 3. 保存当前im-core-server的ip 与 userId之间的绑定关系,设置心跳时间
        stringRedisTemplate.opsForValue().set(ImCoreServerIPRedisKey.ImCoreServerIp + appId + ":" + userId,
                ChannelHandlerContextCache.getImCoreServerAddress() + "%" + userId, ImConstants.DEFAULT_HEART_BEAT_GAP * 2,
                TimeUnit.SECONDS);

        LOGGER.info("login success userId is {} , appId is {}", userId, appId);
        ctx.writeAndFlush(returnMsg);
    }
}